package cn.itcast.streaming.function.window;

import cn.itcast.streaming.bean.ItcastDataPartObj;
import cn.itcast.streaming.utils.DateUtil;
import com.google.common.collect.Lists;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.ArrayList;

//自定义告警规则窗口函数
public class CustomRuleAlarmWindowFunction implements WindowFunction<ItcastDataPartObj, ArrayList<ItcastDataPartObj>,String, TimeWindow> {
    @Override
    public void apply(String s, TimeWindow timeWindow, Iterable<ItcastDataPartObj> iterable, Collector<ArrayList<ItcastDataPartObj>> collector) throws Exception {
        System.out.println("触发窗口["+ DateUtil.timestampToDataStr(timeWindow.getStart())+"-"+DateUtil.timestampToDataStr(timeWindow.getEnd())+"]计算");
        ArrayList<ItcastDataPartObj> objList = Lists.newArrayList(iterable);
        objList.sort((o1, o2) -> {
            if(o1.getTerminalTimeStamp()>o2.getTerminalTimeStamp()){
                return 1;
            }else if(o1.getTerminalTimeStamp()<o2.getTerminalTimeStamp()){
                return -1;
            }else {
                return 0;
            }
        });
        collector.collect(objList);
    }
}
